package com.weepal.pavement.mqtt.demo.config.mqtt;

import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.annotation.IntegrationComponentScan;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.config.EnableIntegration;
import org.springframework.integration.core.MessageProducer;
import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory;
import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;
import org.springframework.messaging.MessageChannel;
import org.springframework.util.StringUtils;

import java.util.UUID;
import java.util.concurrent.atomic.AtomicReference;

@EnableIntegration
@IntegrationComponentScan
@Configuration
public class DefaultMqttConfigration {
    @Autowired
    private DefaultMqttProperties properties;

    @Bean("defatuleMqttInputChannel")
    public MessageChannel defatuleMqttInputChannel() {
        return new DirectChannel();
    }

    /**
     * 连接到MQTT服务器并使用配置的mqtttopic的消息生成器。
     * 并将其放置在mqttmessagerichchannel中。
     */
    @Bean("defaultAdapter")
    public MqttPahoMessageDrivenChannelAdapter defaultAdapter() {
        MqttPahoMessageDrivenChannelAdapter adapter =
                new MqttPahoMessageDrivenChannelAdapter(properties.getAddress(), properties.getClientId() + UUID.randomUUID(),
                        defaultMqttPahoClientFactory());
        adapter.setCompletionTimeout(properties.getTimeOut());
        DefaultPahoMessageConverter converter = new DefaultPahoMessageConverter();
        converter.setPayloadAsBytes(true);
        adapter.setConverter(converter);
        adapter.setQos(properties.getQos());
        adapter.setOutputChannel(defatuleMqttInputChannel());
        MqttClientContainer.putClient(properties.getId(), new AtomicReference<>(null));
        return adapter;
    }

    /**
     * MQTT客户端工厂
     */
    @Bean("defaultMqttPahoClientFactory")
    public MqttPahoClientFactory defaultMqttPahoClientFactory() {
        DefaultMqttPahoClientFactory mqttPahoClientFactory = new DefaultMqttPahoClientFactory();
        mqttPahoClientFactory.setConnectionOptions(defaultMqttConnectOptions());
        return mqttPahoClientFactory;
    }

    /**
     * mqtt配置类
     */
    @Bean("defaultMqttConnectOptions")
    public MqttConnectOptions defaultMqttConnectOptions() {
        MqttConnectOptions options = new MqttConnectOptions();
        if (StringUtils.hasText(properties.getUserName())) {
            options.setUserName(properties.getUserName());
        }
        if (StringUtils.hasText(properties.getPassword())) {
            options.setPassword(properties.getPassword().toCharArray());
        }
        options.setCleanSession(false);
        options.setConnectionTimeout(30);
        options.setKeepAliveInterval(50);
        return options;
    }
}
